package org.uiot.process

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.api.functions.sink.SinkFunction
import org.junit.Test

class EncodingTests extends CodecTestBase {
    @Test
    void testEncodingBySequencePacketBody() {
        given:
        def env = StreamExecutionEnvironment.getExecutionEnvironment()
        env.parallelism = 1

        and:
        CollectSink.values.clear()

        when:
        env.fromElements(new Codec.UnencodedPacket("12345", 01, new Date(), [:], [dataId: 01, deviceId: 0000001, dataTimed: new Date()]))
                .flatMap(new Encoding().cachePackets(packetTraits).cacheRules(codecTraits).cacheSequenceGroups(sequenceTraits))
                .addSink(new CollectSink())

        and:
        env.execute()

        then:
        assert CollectSink.values.first()
        def encodePacket = CollectSink.values.first()
        println encodePacket.getContents().encodeHex()
    }

    // create a testing sink
    private static class CollectSink implements SinkFunction<Codec.EncodedPacket> {

        // must be static
        public static final List<Codec.EncodedPacket> values = []

        @Override
        synchronized void invoke(Codec.EncodedPacket value) throws Exception {
            values.add(value)
        }
    }
}
